#include "Connector.h"
#include <thread>
#include <unistd.h>
#include "ClientInitChannel.h"
#include "DateTime.h"
#include "EventLoop.h"
#include "InnerType.h"
#include "IpV4Address.h"
#include "IpV6Address.h"
#include "PlatLogger.h"
#include "SocketOpt.h"
#include "UnixDomainAddress.h"
#include "Util.h"

namespace xncc {
namespace foundation {
    Connector::Connector(EventLoop* loop, std::shared_ptr<XnccAddress> serverAddr)
        : loop_(loop),
          serverAddr_(std::move(serverAddr)),
          connect_(false),
          state_(DISCONNECTED),
          retryDelayMs_(types::DateTime::MILLS_PER_SEC)
    {
    }
    Connector::~Connector() = default;

    void Connector::start()
    {
        connect_ = true;
        loop_->runInLoop([this]() { startInLoop(false); });
    }

    /// 函数在loop线程中执行，因为client设计为单线程独有，断开重试的时候可以进行sleep操作
    void Connector::startInLoop(bool needSleep)
    {
        if (needSleep) {
            if (retryDelayMs_ > 0) {
                std::this_thread::sleep_for(std::chrono::milliseconds(retryDelayMs_));
            }
        }
        loop_->assertInLoopThread(__FUNCTION__);
        assert(state_ == DISCONNECTED);
        if (connect_) {
            connect();
        }
    }
    void Connector::restart()
    {
        loop_->assertInLoopThread(__FUNCTION__);
        setState(DISCONNECTED);
        connect_ = true;
        startInLoop(true);
    }

    void Connector::stop()
    {
        connect_ = false;
        loop_->runInLoop([this]() { stopInLoop(); });
    }

    void Connector::stopInLoop()
    {
        loop_->assertInLoopThread(__FUNCTION__);
        if (state_ == CONNECTING) {
            setState(DISCONNECTED);
            int sockfd = removeAndResetChannel();
            retry(sockfd);
        }
    }

    void Connector::connect()
    {
        int sockfd = -1;
        int ret    = 0;
        if (serverAddr_->Type() == shared::XNCC_ADDR_TYPE::IPV4_TCP) {
            const IpV4Address::s_ptr v4Addr = std::dynamic_pointer_cast<IpV4Address>(serverAddr_);
            sockfd                          = SocketOpt::createNonblockingOrDie(serverAddr_->family());
            ret = ::connect(sockfd, v4Addr->getSockAddr(), static_cast<socklen_t>(sizeof(struct sockaddr_in)));
        }
        else if (serverAddr_->Type() == shared::XNCC_ADDR_TYPE::IPV6_TCP) {
            const IpV6Address::s_ptr v6Addr = std::dynamic_pointer_cast<IpV6Address>(serverAddr_);
            sockfd                          = SocketOpt::createNonblockingOrDie(serverAddr_->family());
            ret = ::connect(sockfd, v6Addr->getSockAddr(), static_cast<socklen_t>(sizeof(struct sockaddr_in6)));
        }
        else if (serverAddr_->Type() == shared::XNCC_ADDR_TYPE::UNIX_DOMAIN) {
            const UnixDomainAddress::s_ptr udsAddr = std::dynamic_pointer_cast<UnixDomainAddress>(serverAddr_);
            sockfd                         = SocketOpt::createNonblockingOrDie(serverAddr_->family(), IPPROTO_IP);
            struct sockaddr_un clientAddr_ = {};
            clientAddr_.sun_family         = AF_UNIX;
            std::string pidStr             = "/tmp/xncc_client_" + std::to_string(getpid());
            if (shared::FileBusi::exists(pidStr)) {
                shared::FileBusi::remove(pidStr);
            }
            strncpy(clientAddr_.sun_path, pidStr.data(), pidStr.size());
            ret = ::bind(sockfd, SocketOpt::sockaddr_cast(&clientAddr_),
                         static_cast<socklen_t>(sizeof(struct sockaddr_un)));
            if (ret < 0) {
                throw shared::system_exception("bind socket failed");
            }
            ret = ::connect(sockfd, udsAddr->getSockAddr(), static_cast<socklen_t>(sizeof(struct sockaddr_un)));
        }
        int savedErrno = (ret == 0) ? 0 : errno;
        switch (savedErrno) {
            case 0:
            case EINPROGRESS:
            case EINTR:
            case EISCONN: {
                connecting(sockfd);
            } break;

                // case ENOENT:  // domain socket error
                // case EAGAIN:
                // case EADDRINUSE:
                // case EADDRNOTAVAIL:
                // case ECONNREFUSED:
                // case ENETUNREACH:

            default: {  // error continue
                SYSTEM_ERROR << "connect error  try again "_s;
                retry(sockfd);
            } break;

                // case EACCES:
                // case EPERM:
                // case EAFNOSUPPORT:
                // case EALREADY:
                // case EBADF:
                // case EFAULT:
                // case ENOTSOCK:
                //     SYSTEM_ERROR << "connect error in Connector::startInLoop"_s;
                //     SocketOpt::close(sockfd);
                //     break;

                // default:
                //     SYSTEM_ERROR << "Unexpected error in Connector::startInLoop "_s;
                //     SocketOpt::close(sockfd);
                //     // connectErrorCallback_();
                //    break;
        }
    }

    void Connector::connecting(int sockfd)
    {
        setState(CONNECTING);
        assert(!channel_);
        /// 这个channel只是用来判断连接建立成功的，不做收发数据使用
        channel_ = std::make_shared<ClientInitChannel>(loop_, sockfd);
        channel_->init();
        /// 注册了写回掉，如果真的连接成功了，fd就可写了，就会触发写回调，那么就证明连接真正成功了
        /// 只靠connect返回判断连接成功是不完全靠谱的
        channel_->setWriteCallback([this]() { handleWrite(); });
        channel_->setErrorCallback([this](const std::string& reason) { handleError(); });
        channel_->enableWriting();
    }

    void Connector::handleWrite()
    {
        if (state_ == CONNECTING) {
            int sockfd = removeAndResetChannel();
            int err    = SocketOpt::getSocketError(sockfd);
            if (err != 0) {
                TRACE_ERROR << ("Connector::handleWrite - SO_ERROR =  ") << err << " " << shared::strerror_tl(err);
                retry(sockfd);
            }
            else if (SocketOpt::isSelfConnect(sockfd)) {
                TRACE_ERROR << ("Connector::handleWrite - Self connect");
                retry(sockfd);
            }
            else {
                setState(CONNECTED);
                if (connect_) {
                    newConnectionCallback_(sockfd);
                }
                else {
                    SocketOpt::close(sockfd);
                }
            }
        }
        else {
            // what happened?
            assert(state_ == DISCONNECTED);
        }
    }

    void Connector::shutdown(std::shared_ptr<Channel> ch)
    {
        loop_->runInLoop([this, ch]() { shutdownInLoop(ch); });
    }

    void Connector::shutdownInLoop(const std::shared_ptr<Channel>& channel)
    {
        loop_->assertInLoopThread(__FUNCTION__);
        if (!channel->isWriting()) {
            SocketOpt::shutdownWrite(channel->fd());
        }
    }

    int Connector::removeAndResetChannel()
    {
        channel_->disableAll();
        channel_->remove();
        int sockfd = channel_->fd();
        // Can't reset channel_ here, because we are inside Channel::handleEvent
        // 直接释放指针  loop有可能还在使用中
        loop_->runInLoop([this]() { resetChannel(); });

        return sockfd;
    }

    void Connector::resetChannel() { channel_.reset(); }

    void Connector::handleError()
    {
        TRACE_ERROR << ("Connector::handleError state= ") << state_;
        if (state_ == CONNECTING) {
            int sockfd = removeAndResetChannel();
            int err    = SocketOpt::getSocketError(sockfd);
            TRACE_ERROR << ("SO_ERROR = ") << err << " " << shared::strerror_tl(err);
            retry(sockfd);
        }
    }

    void Connector::retry(int sockfd)
    {
        SocketOpt::close(sockfd);
        setState(DISCONNECTED);
        if (connect_) {
            TRACE_INFO << ("Connector::retry-Retry connecting to") << serverAddr_->toStr() << " after " << retryDelayMs_
                       << " milliseconds";
            loop_->runInLoop([this]() { startInLoop(true); });

            // 定时重连
        }
    }
}  // namespace foundation
}  // namespace xncc